#include "app_pool.h"
#include <mqueue.h>
#include <pthread.h>
#include <stdlib.h>
#include <inttypes.h>
#include "log/log.h"

typedef struct 
{
    TaskFunc func;
    void *arg;
} Task;

#define MSG_LEN sizeof(Task)
#define MSG_NAME "/task_queue"

static mqd_t mq = -1;
static pthread_t *threads = NULL;
static int thread_count = 0;

void *app_pool_worker(void *arg)
{
    //需要从消息队列中读取Task结构体，并执行函数
    intptr_t worker_id = (intptr_t)arg;
    log_info("worker %" PRIiPTR "created", worker_id);
    Task task;
    while(1)
    {
        int ret = mq_receive(mq, (char*)&task, MSG_LEN, NULL);
        if (ret < 0)
        {
            log_error("Receive task failed");
            continue;
        }
        task.func(task.arg);

    }
}

/**
 * @brief 初始化线程池
 * @param num_threads 线程池中线程数量
 * @return 0 成功，-1 失败
 */
int app_pool_init(int num_threads)
{
    //1.创建线程池的队列
    struct mq_attr attr =
    {
        .mq_maxmsg = 10,
        .mq_msgsize = MSG_LEN,
    };
    mq = mq_open(MSG_NAME, O_RDWR | O_CREAT, 0644, &attr);
    if (mq < 0)
    {
        log_error("create mq failed");
        goto MQ_OPEN_FAIL;
    }

    //2.创建线程池中的后台线程
    thread_count = num_threads;
    threads = malloc(sizeof(pthread_t) * num_threads);
    if (!threads)
    {
        log_error("Pool allocate memory failed");
        goto POOL_ALLOC_FAIL;
    }
    intptr_t i = 0;
    for(i = 0; i < num_threads; i++)
    {
        int ret = pthread_create(threads +i, NULL, app_pool_worker, (void *)i);
        if (ret != 0)
        {
            break;
        }
    }
    if (i < num_threads)
    {
        log_error("Created %d threads, failed",i);
        goto THREAD_CREATE_FAIL;
    }
    log_info("Pool_init success");

    return 0;

THREAD_CREATE_FAIL:
    //1.关闭所有已经创建的线程
    for (int j = 0; j < i; j++)
    {
        pthread_cancel(threads[j]);
        pthread_join(threads[j], NULL);
    }
    //2.清理threads变量
    free(threads);

POOL_ALLOC_FAIL:
    mq_unlink(MSG_NAME);

MQ_OPEN_FAIL:
    return -1;

}

/**
 * @brief 向线程池提交任务、
 * @param func 任务句柄
 * @param arg 任务参数
 * @return 0 成功，-1 失败
 */
int app_pool_submit(TaskFunc func, void *arg)
{
    //1. 向线程池队列中提交任务
    Task task = 
    {
        .func = func,
        .arg = arg,
    };
    log_debug("Submiting task %p", func);
    return mq_send(mq, (char*)&task, MSG_LEN, 0);
}

/**
 * @brief 销毁线程池
 */
void app_pool_deinit()
{
    //1.关闭线程池后台的所有线程
    for (int i = 0; i < thread_count; i++)
    {
        pthread_cancel(threads[i]);
        pthread_join(threads[i], NULL);
    }
    free(threads);

    //2.关闭线程池队列
    mq_unlink(MSG_NAME);
    log_info("Pool_deinit success");
}










